package org.opens.netty.chatServer.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.Promise;
import lombok.extern.slf4j.Slf4j;
import org.opens.netty.chatServer.client.handle.RpcResponseMessageHandle;
import org.opens.netty.chatServer.message.RpcRequestMessage;
import org.opens.netty.chatServer.protocol.MessageCodecSharable;
import org.opens.netty.chatServer.protocol.ProtocolFrameDecoder;
import org.opens.netty.chatServer.protocol.SequenceIdGenerator;
import org.opens.netty.chatServer.server.service.HelloService;

import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;

@Slf4j
public class RpcClientManager {

    private static Channel channel = null;

    public static void main(String[] args) throws InterruptedException {
        HelloService helloServiceProxy = getProxyService(HelloService.class);
        System.out.println(helloServiceProxy.sayHello("zhangsan"));
        System.out.println(helloServiceProxy.sayHello("lisi"));
    }

    // 创建代理类
    public static <T> T getProxyService(Class<T> interfaceClass) {
        Class<?>[] interfaces = new Class[]{interfaceClass};
        Object proxyObject = Proxy.newProxyInstance(interfaceClass.getClassLoader(), interfaces, (proxy, method, args) -> {
            // 1. 将方法调用转换为消息对象
            int sequenceId = SequenceIdGenerator.nexId();
            RpcRequestMessage rpcRequestMessage = new RpcRequestMessage(
                    sequenceId,
                    interfaceClass.getName(),
                    method.getName(),
                    method.getReturnType(),
                    method.getParameterTypes(),
                    args
            );

            // 2. 将消息发出去
            getChannel().writeAndFlush(rpcRequestMessage);

            // 3. 准备一个Promise对象, 来接受结果
            // 构造的参数是指定一个线程用来异步接受结果
            DefaultPromise<Object> promise = new DefaultPromise<>(getChannel().eventLoop());
            RpcResponseMessageHandle.PROMISE_MAP.put(sequenceId, promise);

            // 4. 等待Promise的结果
            // await不论Promise中的结果是否成功不会抛出异常, 而sync会在失败时抛出异常; 这两个方法都是阻塞方法
            promise.await();

            System.out.println("当前线程: " + Thread.currentThread());

            // 5. 调用正常返回结果, 调用失败返回异常信息
            if (promise.isSuccess()) {
                return promise.getNow();
            } else {
                throw new RuntimeException(promise.cause());
            }
        });
        return (T) proxyObject;
    }

    /**
     * 单例模式获取channel, 线程安全
     */
    public static Channel getChannel() {
        if (channel != null) {
            return channel;
        }
        // 安全保护
        synchronized (RpcClientManager.class) {
            if (channel != null) {
                return channel;
            }
            initChannel();
            return channel;
        }
    }

    /**
     * 初始化channel
     */
    private static void initChannel() {
        NioEventLoopGroup worker = new NioEventLoopGroup();

        LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
        MessageCodecSharable MESSAGE_CODEC_SHARABLE = new MessageCodecSharable();

        RpcResponseMessageHandle rpcResponseMessageHandle = new RpcResponseMessageHandle();


        Bootstrap bootstrap = new Bootstrap();
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.group(worker);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new ProtocolFrameDecoder());
                ch.pipeline().addLast(LOGGING_HANDLER);
                ch.pipeline().addLast(MESSAGE_CODEC_SHARABLE);
                ch.pipeline().addLast(rpcResponseMessageHandle);
            }
        });

        try {
            ChannelFuture channelFuture = bootstrap.connect(new InetSocketAddress("127.0.0.1", 8080)).sync();
            channel = channelFuture.channel();
            // 异步方式关闭, sync方法会阻塞
            channel.closeFuture().addListener(new GenericFutureListener() {
                @Override
                public void operationComplete(Future future) throws Exception {
                    worker.shutdownGracefully();
                    // 清空channel引用
                    channel = null;
                }
            });
        } catch (InterruptedException e) {
            log.error("client error: ", e);
        }
    }
}
